package com.pengjunlee.hive.controller;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping("/spark")
public class SparkKafkaStreamingController {
	
	@SuppressWarnings("unused")
	private static final long serialVersionUID = 1L;

	/**
	 * 从kafka读取流数据
	 * @throws InterruptedException 
	 */
	@SuppressWarnings("unused")
	@RequestMapping("/streaming")
	public void streaming() throws InterruptedException {
		SparkConf sparkConf  = new SparkConf();
		sparkConf.setMaster("local[4]");
		sparkConf.setAppName("SparkStreamingFromkafka");
		sparkConf.set("spark.streaming.stopGracefullyOnShutdown","true");
 		sparkConf.set("spark.default.parallelism", "6");
		JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, new Duration(10000));
		
		//kafka配置信息
		Map<String, Object> kafkaParams = new HashMap<String, Object>();
		//Kafka服务监听端口,多个可用ip可用","隔开
		kafkaParams.put("bootstrap.servers", "10.8.23.4:9092");
		//指定kafka输出key的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        //指定kafka输出value的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        //消费者ID，随意指定
        kafkaParams.put("group.id", "sparkStreaming");
        //指定从latest(最新,其他版本的是largest这里不行)还是smallest(最早)处开始读取数据
        kafkaParams.put("auto.offset.reset", "latest");
        //如果true,consumer定期地往zookeeper写入每个分区的offset
        kafkaParams.put("enable.auto.commit", false);
        //要监听的Topic，可以同时监听多个
        Collection<String> topics = Arrays.asList("test");
        
        JavaInputDStream<ConsumerRecord<String, String>> javaInputDStream = KafkaUtils.createDirectStream(
        		streamingContext, 
        		LocationStrategies.PreferConsistent(), 
        		ConsumerStrategies.Subscribe(topics, kafkaParams));
        
//        javaInputDStream.mapToPair(
//        	    new PairFunction<ConsumerRecord<String, String>, String, String>() {
//        	    @Override
//        	    public Tuple2<String, String> call(ConsumerRecord<String, String> record) {
//        	        return new Tuple2<>(record.key(), record.value());
//        	    }
//    	});
        
         streamingContext.start();
         streamingContext.awaitTermination();
	}
}